package com.example.factory;

import com.example.transport.ServerTransportService;
import com.exmaple.dto.RpcPing;
import com.exmaple.dto.RpcResponse;
import com.exmaple.dto.RpcRequest;
import com.exmaple.loadBalance.RpcLoadBalance;

import java.util.Map;
import java.util.concurrent.*;
import java.util.logging.Logger;

public class SysAdminServerFactory extends AbstractServerFactory implements ServerTransportService {

    Logger logger = Logger.getLogger(getClass().getName());

    private Map<String, RpcPing> pingMap = new ConcurrentHashMap<>();

    private Thread pingCheckThread;

    private ExecutorService executorService = null;

    private int corePoolSize;

    private int maximumPoolSize;

    private long keepAliveTime;

    public SysAdminServerFactory(int corePoolSize, int maximumPoolSize, long keepAliveTime) {
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.keepAliveTime = keepAliveTime;
    }

    @Override
    public void init() {
        this.executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS, new LinkedBlockingDeque());
    }

    @Override
    public void start() {
        this.pingCheckThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    for (String key : pingMap.keySet()) {
                        RpcPing rpcPing = pingMap.get(key);
                        if (rpcPing.getNow() + FIXED_TIME < System.currentTimeMillis()) {
                            logger.warning(">>>>>>> 节点: {} 已掉线 >>>>>>>" + rpcPing.getUrl());
                            pingMap.remove(key);
                        }
                    }
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                }
            }
        });
        this.pingCheckThread.setDaemon(true);
        this.pingCheckThread.setName("pingCheckThread");
        this.pingCheckThread.start();
        logger.info(">>>>>>>>> pingCheckThread started  >>>>>>>>>");
    }

    @Override
    public void stop() {
        if (this.pingCheckThread.isAlive()) {
            this.pingCheckThread.stop();
        }
        this.executorService.shutdown();
    }

    @Override
    public RpcResponse handler(RpcRequest rpcRequest) {
        RpcLoadBalance rpcLoadBalance = rpcRequest.getLoadBalanceEnum().getRpcLoadBalance();
        RpcPing rpcPing = rpcLoadBalance.route(pingMap, rpcRequest);
        if (rpcPing == null) {
            return RpcResponse.error(rpcRequest, "未找到执行节点");
        } else {
            return handler(rpcPing.getUrl(), rpcRequest);
        }
    }

    @Override
    public RpcResponse handler(RpcPing rpcPing) {
        this.pingMap.put(rpcPing.getUrl(), rpcPing);
        return new RpcResponse();
    }

    @Override
    public RpcResponse handler(String targetAddress, RpcRequest rpcRequest) {
        rpcRequest.setTargetId(targetAddress);
        RpcResponse rpcResponse = null;
        try {
            Future<RpcResponse> responseFuture = this.executorService.submit(new Callable<RpcResponse>() {
                @Override
                public RpcResponse call() throws Exception {
                    return handlerHTTP(HTTP + targetAddress + PREF + Executor, rpcRequest);
                }
            });
            rpcResponse = responseFuture.get(FIXED_TIME, TimeUnit.SECONDS);
        } catch (Exception e) {
            rpcResponse.setSuccess(Boolean.FALSE);
            rpcResponse.setErrInfo(e.getMessage());
        }
        return rpcResponse;
    }
}
